package cn.edu.ccnu.nlp;

import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;

public class Search {
	
	public static Configuration configuration;
	public static HTablePool pool;
	public static HTable news_stat;
	public static HTable blog_stat;
	public static HTable bbs_stat;
	static {
		configuration = HBaseConfiguration.create();
		
		try{
			pool = new HTablePool(configuration, 1000);
			news_stat = (HTable)pool.getTable("news_stat");
			//blog_stat = (HTable)pool.getTable("blog_stat");
			//bbs_stat = (HTable)pool.getTable("bbs_stat");
		}catch(Exception e){
			e.getStackTrace();
		}
	}
	
	
	/**
	 * 统计某个年份中每个月的数据量
	 * @param tableName
	 */
	public static HashMap<String,Integer> QueryByYearsAndMonths(HTable table) {

		int sum = 0;
		String year = null;
		ResultScanner rs = null;
		HashMap<String,Integer> monthNum = new HashMap<String,Integer>();
		SimpleDateFormat sdf=new SimpleDateFormat("yyyy");                     //获取当前年份，以便于遍历2005年到当前的年份
		String current_year=sdf.format(new java.util.Date());  
		String[] months = {"01","02","03","04","05","06","07","08","09","10","11","12"};
		try { 
				for(int i = 2005; i < Integer.parseInt(current_year); i++){
					year = ""+i;
					for(int j = 0; j < months.length; j++){
						Filter filter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(year+"-"+months[j]+"-.*")); // 查询某年某月的记录数
						Scan s = new Scan();
						s.setFilter(filter);
						rs = table.getScanner(s);
						for (Result r : rs) {
							for (KeyValue keyValue : r.raw()) {
									sum += Integer.parseInt(new String(keyValue.getValue()));
							}
						}
						System.out.println("year-month:"+year+"-"+months[j]+":sum:"+sum);
						monthNum.put(year+"-"+months[j], sum);
						sum = 0;
					}
				}
		} catch (Exception e) {
			e.printStackTrace();
		}
		finally{
			rs.close();
		}
		return monthNum;
	}

	public static void main(String[] args){
		QueryByYearsAndMonths(news_stat);
	}
}
